/**
 * 
 */
package com.zhb.cloud.core.distribute.zookeeper.lock;

import java.util.concurrent.TimeUnit;

import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zhb.cloud.core.distribute.lock.DistributedLock;
import com.zhb.cloud.core.distribute.lock.DistributedLockException;

/**
 * @ClassName: ZooKeeperDistributedLock
 * @author: zhb
 * @date: 2017年8月4日 上午11:08:01
 * @Description: zookeeper 分布式锁实现
 *
 */
public class ZooKeeperDistributedLock implements DistributedLock, ConnectionStateListener {
	private static final Logger LOG = LoggerFactory.getLogger(ZooKeeperDistributedLock.class);

	public static final String NAMESPACE = "/px/locks";

	/**
	 * initial amount of time to wait between retries in milliseconds
	 */
	private static final int ZOOKEEPER_RETRY_INTERVAL = 1 * 1000;

	/**
	 * retry timeout in milliseconds
	 */
	private static final int ZOOKEEPER_RETRY_TIMEOUT = 10 * 1000;

	/**
	 * rehold the lock if connect lost 如果断开连接重新获得锁
	 */
	private static final boolean RECONNECT_FLAG = true;

	/**
	 * zookeeper client 客户端
	 */
	private CuratorFramework zkClient;

	/**
	 * Distributed zookeeper lock 分布式zookeeper锁
	 */
	private InterProcessLock interProcessLock;

	/**
	 * lock namespace
	 */
	@SuppressWarnings("unused")
	private String namespace;

	/**
	 * The lock path
	 */
	private String lockPath;

	/**
	 * @param client
	 *            Curator client
	 * @param path
	 *            the lock path, such as /example/lock
	 * @param namespace
	 *            the lock namespace
	 */
	public ZooKeeperDistributedLock(CuratorFramework client, String path, String namespace) {
		initDistributedLock(client, path, namespace);
	}

	/**
	 * @param connectString
	 *            Curator client
	 * @param lockPath
	 *            the lock path, such as /example/lock
	 * @param namespace
	 *            the lock namespace
	 * @DESC 获得zk客户端连接,并初始化zk分布式锁
	 */
	public ZooKeeperDistributedLock(String connectString, String lockPath, String namespace) {
		LOG.info("------------ 当前zookeeper 地址：{}, 锁路径：{} ------------", connectString, lockPath);
		CuratorFramework client = ZooKeeperClientFactory.createZooKeeperClient(connectString, namespace);
		initDistributedLock(client, lockPath, namespace);
	}

	/**
	 * Construct with default namespace: {@value #NAMESPACE}
	 *
	 * @param connectString
	 *            Curator client
	 * @param lockPath
	 *            the lock path, such as /example/lock
	 */
	public ZooKeeperDistributedLock(String connectString, String lockPath) {
		this(connectString, lockPath, NAMESPACE);
	}

	/**
	 * Initialize distributed lock
	 *
	 * @param client
	 *            Curator client
	 * @param lockPath
	 *            the path to lock
	 * @param namespace
	 *            the lock namespace
	 */
	private void initDistributedLock(CuratorFramework client, String lockPath, String namespace) {
		if (client == null) {
			throw new DistributedLockException("no zookeeper client found");
		}
		this.interProcessLock = new InterProcessMutex(client, lockPath);
		this.zkClient = client;
		this.namespace = namespace;
		this.lockPath = namespace + lockPath;
		LOG.info("Initialize zookeeper lock instance, path={} zkclient={}", this.lockPath, client);
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see cn.px.hq.server.locks.DistributedLock#lock()
	 */
	@Override
	public void lock() throws DistributedLockException {
		try {
			this.interProcessLock.acquire();
			registerStateListenable();
			LOG.info("{} acquire lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new DistributedLockException("failed to acquire the lock", e);
		}
	}

	/**
	 * (non-Javadoc)
	 * 
	 * @see cn.px.hq.server.locks.DistributedLock#lock(long, TimeUnit)
	 */
	@Override
	public void lock(long time, TimeUnit unit) throws DistributedLockException {
		try {
			// FIXME: lock expiration
			this.interProcessLock.acquire();
			registerStateListenable();
			LOG.info("{} acquire lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new DistributedLockException("failed to acquire the lock", e);
		}
	}

	@Override
	public void unlock() throws DistributedLockException {
		try {
			this.interProcessLock.release();
			unregisterStateListenable();
			LOG.info("{} release lock: {}", Thread.currentThread(), lockPath);
		} catch (Exception e) {
			throw new DistributedLockException("failed to release the lock", e);
		}
	}

	@Override
	public boolean tryLock() throws DistributedLockException {
		// HACK(subo): wait for 1 millisecond if the lock was not acquired
		return tryLock(1, TimeUnit.MILLISECONDS);
	}

	@Override
	public boolean tryLock(long time, TimeUnit unit) throws DistributedLockException {
		boolean result = false;
		try {
			result = this.interProcessLock.acquire(time, unit);
			if (result) {
				registerStateListenable();
			}
		} catch (Exception e) {
			throw new DistributedLockException("failed to try acquire the lock", e);
		}
		LOG.info("{} try lock: {} result: {}", Thread.currentThread(), this.lockPath, result);
		return result;
	}

	@Override
	public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) {
		// throw new RuntimeException("method unsupported");
		return tryLock(waitTime, unit);
	}

	@Override
	public boolean isLocked() throws DistributedLockException {
		// FIXME: remote lock check
		return this.interProcessLock.isAcquiredInThisProcess();
	}

	@Override
	public void stateChanged(CuratorFramework client, ConnectionState newState) {
		if (newState != ConnectionState.LOST)
			return;
		if (!this.interProcessLock.isAcquiredInThisProcess())
			return;

		// rehold lock after connect lost
		while (true) {
			try {
				if (client.getZookeeperClient().blockUntilConnectedOrTimedOut()) {
					this.interProcessLock.acquire(ZOOKEEPER_RETRY_TIMEOUT, TimeUnit.MILLISECONDS);
					break;
				}
				Thread.sleep(ZOOKEEPER_RETRY_INTERVAL);
			} catch (Exception e) {
				LOG.warn(ExceptionUtils.getStackTrace(e));
				break;
			}
		}
	}

	/**
	 * register state listenable
	 */
	private void registerStateListenable() {
		if (!RECONNECT_FLAG)
			return;

		this.zkClient.getConnectionStateListenable().addListener(this);
	}

	/**
	 * unregister state listenable
	 */
	private void unregisterStateListenable() {
		if (!RECONNECT_FLAG)
			return;

		if (!this.interProcessLock.isAcquiredInThisProcess()) {
			this.zkClient.getConnectionStateListenable().removeListener(this);
		}
	}
}
